package cn.azzhu.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @author azzhu
 * @create 2020-09-19 13:59:17
 */
public class FlinkUtilsV1 {

    private static final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    /**
     * 从kafka中读取数据
     * @param args
     * @param simpleStringSchema
     * @return
     */
    public static DataStream<String> createKafkaStream(String[] args, SimpleStringSchema simpleStringSchema) {
        String topic = args[0];
        String groupId = args[1];
       // String brokerList = args[2];
        //kafkaSource
        final Properties props = new Properties();
        //指定kafka的broker地址
        props.setProperty("bootstrap.servers", "hadoop105:9092,hadoop106:9092,hadoop107:9092");
        //指定组ID
        props.setProperty("group.id", groupId);
        //如果没有指定偏移量，第一次从最开始消费
        props.setProperty("auto.offset.reset","earliest");
        //kafka的消费者不自动提交偏移量
//        props.setProperty("enable.auto.commit","false");

        final FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>(
                topic,
                simpleStringSchema,
                props
        );

        //source
        final DataStream<String> lines = env.addSource(kafkaSource);
        return lines;
    }

    public static StreamExecutionEnvironment getEnv() {
        return env;
    }
}
